// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//

// Object model for scalar (non-Array) values. Not intended for use with large
// amounts of data

#pragma once

#include <iosfwd>
#include <memory>
#include <ratio>
#include <string>
#include <string_view>
#include <utility>
#include <vector>

#include <nebula/core/compare.h>
#include <nebula/core/extension_type.h>

#include <turbo/utility/status.h>
#include <nebula/types/type.h>
#include <nebula/types/type_fwd.h>
#include <nebula/types/type_traits.h>
#include <turbo/meta/compare.h>
#include <nebula/numeric/decimal.h>
#include <turbo/base/macros.h>
#include <nebula/core/visit_type_inline.h>

namespace nebula {

    class Array;

    /// \brief Base class for scalar values
    ///
    /// A Scalar represents a single value with a specific DataType.
    /// Scalars are useful for passing single value inputs to compute functions,
    /// or for representing individual array elements (with a non-trivial
    /// wrapping cost, though).
    struct TURBO_EXPORT Scalar : public std::enable_shared_from_this<Scalar>,
                                 public turbo::EqualityComparable<Scalar> {
        virtual ~Scalar() = default;

        /// \brief The type of the scalar value
        std::shared_ptr<DataType> type;

        /// \brief Whether the value is valid (not null) or not
        bool is_valid = false;

        bool equals(const Scalar &other,
                    const EqualOptions &options = EqualOptions::defaults()) const;

        bool approx_equals(const Scalar &other,
                          const EqualOptions &options = EqualOptions::defaults()) const;

        struct TURBO_EXPORT Hash {
            size_t operator()(const Scalar &scalar) const { return scalar.hash(); }

            size_t operator()(const std::shared_ptr<Scalar> &scalar) const {
                return scalar->hash();
            }
        };

        size_t hash() const;

        std::string to_string() const;

        /// \brief Perform cheap validation checks
        ///
        /// This is O(k) where k is the number of descendents.
        ///
        /// \return turbo::Status
        turbo::Status validate() const;

        /// \brief Perform extensive data validation checks
        ///
        /// This is potentially O(k*n) where k is the number of descendents and n
        /// is the length of descendents (if list scalars are involved).
        ///
        /// \return turbo::Status
        turbo::Status validate_full() const;

        static turbo::Result<std::shared_ptr<Scalar>> parse(const std::shared_ptr<DataType> &type,
                                                            std::string_view repr);

        // TODO(bkietz) add compute::CastOptions
        turbo::Result<std::shared_ptr<Scalar>> cast_to(std::shared_ptr<DataType> to) const;

        /// \brief Apply the ScalarVisitor::Visit() method specialized to the scalar type
        turbo::Status Accept(ScalarVisitor *visitor) const;

        /// \brief EXPERIMENTAL Enable obtaining shared_ptr<Scalar> from a const
        /// Scalar& context.
        std::shared_ptr<Scalar> get_shared_ptr() const {
            return const_cast<Scalar *>(this)->shared_from_this();
        }

    protected:
        Scalar(std::shared_ptr<DataType> type, bool is_valid)
                : type(std::move(type)), is_valid(is_valid) {}
    };

    TURBO_EXPORT void print_to(const Scalar &scalar, std::ostream *os);

    /// \defgroup concrete-scalar-classes Concrete Scalar subclasses
    ///
    /// @{

    /// \brief A scalar value for NullType. Never valid
    struct TURBO_EXPORT NullScalar : public Scalar {
    public:
        using TypeClass = NullType;

        NullScalar() : Scalar{null(), false} {}
    };

    /// @}

    namespace internal {

        constexpr auto kScalarScratchSpaceSize = sizeof(int64_t) * 2;

        template<typename Impl>
        struct TURBO_EXPORT array_span_fill_from_scalar_scratch_space {
            //  16 bytes of scratch space to enable ArraySpan to be a view onto any
            //  Scalar- including binary scalars where we need to create a buffer
            //  that looks like two 32-bit or 64-bit offsets.
            alignas(int64_t) mutable uint8_t scratch_space_[kScalarScratchSpaceSize];

        private:
            template<typename... Args>
            explicit array_span_fill_from_scalar_scratch_space(Args &&... args) {
                Impl::FillScratchSpace(scratch_space_, std::forward<Args>(args)...);
            }

            array_span_fill_from_scalar_scratch_space() = delete;

            friend Impl;
        };

        struct TURBO_EXPORT PrimitiveScalarBase : public Scalar {
            explicit PrimitiveScalarBase(std::shared_ptr<DataType> type)
                    : Scalar(std::move(type), false) {}

            using Scalar::Scalar;

            /// \brief Get a const pointer to the value of this scalar. May be null.
            virtual const void *data() const = 0;

            /// \brief Get an immutable view of the value of this scalar as bytes.
            virtual std::string_view view() const = 0;
        };

        template<typename T, typename CType = typename T::c_type>
        struct TURBO_EXPORT PrimitiveScalar : public PrimitiveScalarBase {
            using PrimitiveScalarBase::PrimitiveScalarBase;
            using TypeClass = T;
            using value_type = CType;

            // Non-null constructor.
            PrimitiveScalar(value_type value, std::shared_ptr<DataType> type)
                    : PrimitiveScalarBase(std::move(type), true), value(value) {}

            explicit PrimitiveScalar(std::shared_ptr<DataType> type)
                    : PrimitiveScalarBase(std::move(type), false) {}

            value_type value{};

            const void *data() const override { return &value; }

            std::string_view view() const override {
                return std::string_view(reinterpret_cast<const char *>(&value), sizeof(value_type));
            };
        };

    }  // namespace internal

    /// \addtogroup concrete-scalar-classes Concrete Scalar subclasses
    ///
    /// @{

    struct TURBO_EXPORT BooleanScalar : public internal::PrimitiveScalar<BooleanType> {
        using Base = internal::PrimitiveScalar<BooleanType, bool>;
        using Base::Base;

        explicit BooleanScalar(bool value) : Base(value, boolean()) {}

        BooleanScalar() : Base(boolean()) {}
    };

    template<typename T>
    struct NumericScalar : public internal::PrimitiveScalar<T> {
        using Base = typename internal::PrimitiveScalar<T>;
        using Base::Base;
        using TypeClass = typename Base::TypeClass;
        using value_type = typename Base::value_type;

        explicit NumericScalar(value_type value)
                : Base(value, TypeTraits<T>::type_singleton()) {}

        NumericScalar() : Base(TypeTraits<T>::type_singleton()) {}
    };

    struct TURBO_EXPORT Int8Scalar : public NumericScalar<Int8Type> {
        using NumericScalar<Int8Type>::NumericScalar;
    };

    struct TURBO_EXPORT Int16Scalar : public NumericScalar<Int16Type> {
        using NumericScalar<Int16Type>::NumericScalar;
    };

    struct TURBO_EXPORT Int32Scalar : public NumericScalar<Int32Type> {
        using NumericScalar<Int32Type>::NumericScalar;
    };

    struct TURBO_EXPORT Int64Scalar : public NumericScalar<Int64Type> {
        using NumericScalar<Int64Type>::NumericScalar;
    };

    struct TURBO_EXPORT UInt8Scalar : public NumericScalar<UInt8Type> {
        using NumericScalar<UInt8Type>::NumericScalar;
    };

    struct TURBO_EXPORT UInt16Scalar : public NumericScalar<UInt16Type> {
        using NumericScalar<UInt16Type>::NumericScalar;
    };

    struct TURBO_EXPORT UInt32Scalar : public NumericScalar<UInt32Type> {
        using NumericScalar<UInt32Type>::NumericScalar;
    };

    struct TURBO_EXPORT UInt64Scalar : public NumericScalar<UInt64Type> {
        using NumericScalar<UInt64Type>::NumericScalar;
    };

    struct TURBO_EXPORT Fp16Scalar : public NumericScalar<Fp16Type> {
        using NumericScalar<Fp16Type>::NumericScalar;
    };

    struct TURBO_EXPORT Fp32Scalar : public NumericScalar<Fp32Type> {
        using NumericScalar<Fp32Type>::NumericScalar;
    };

    struct TURBO_EXPORT Fp64Scalar : public NumericScalar<Fp64Type> {
        using NumericScalar<Fp64Type>::NumericScalar;
    };

    struct TURBO_EXPORT BaseBinaryScalar : public internal::PrimitiveScalarBase {
        using value_type = std::shared_ptr<Buffer>;

        // The value is not supposed to be modified after construction, because subclasses have
        // a scratch space whose content need to be kept consistent with the value. It is also
        // the user of this class's responsibility to ensure that the buffer is not written to
        // accidentally.
        const std::shared_ptr<Buffer> value = nullptr;

        const void *data() const override {
            return value ? reinterpret_cast<const void *>(value->data()) : nullptr;
        }

        std::string_view view() const override {
            return value ? std::string_view(*value) : std::string_view();
        }

        explicit BaseBinaryScalar(std::shared_ptr<DataType> type)
                : internal::PrimitiveScalarBase(std::move(type)) {}

        BaseBinaryScalar(std::shared_ptr<Buffer> value, std::shared_ptr<DataType> type)
                : internal::PrimitiveScalarBase{std::move(type), true}, value(std::move(value)) {}

        BaseBinaryScalar(std::string s, std::shared_ptr<DataType> type);
    };

    struct TURBO_EXPORT BinaryScalar
            : public BaseBinaryScalar,
              private internal::array_span_fill_from_scalar_scratch_space<BinaryScalar> {
        using TypeClass = BinaryType;
        using array_span_fill_from_scalar_scratch_space =
                internal::array_span_fill_from_scalar_scratch_space<BinaryScalar>;

        explicit BinaryScalar(std::shared_ptr<DataType> type)
                : BaseBinaryScalar(std::move(type)),
                  array_span_fill_from_scalar_scratch_space(this->value) {}

        BinaryScalar(std::shared_ptr<Buffer> value, std::shared_ptr<DataType> type)
                : BaseBinaryScalar(std::move(value), std::move(type)),
                  array_span_fill_from_scalar_scratch_space(this->value) {}

        BinaryScalar(std::string s, std::shared_ptr<DataType> type)
                : BaseBinaryScalar(std::move(s), std::move(type)),
                  array_span_fill_from_scalar_scratch_space(this->value) {}

        explicit BinaryScalar(std::shared_ptr<Buffer> value)
                : BinaryScalar(std::move(value), binary()) {}

        explicit BinaryScalar(std::string s) : BinaryScalar(std::move(s), binary()) {}

        BinaryScalar() : BinaryScalar(binary()) {}

    private:
        static void FillScratchSpace(uint8_t *scratch_space,
                                     const std::shared_ptr<Buffer> &value);

        friend ArraySpan;
        friend array_span_fill_from_scalar_scratch_space;
    };

    struct TURBO_EXPORT StringScalar : public BinaryScalar {
        using BinaryScalar::BinaryScalar;
        using TypeClass = StringType;

        explicit StringScalar(std::shared_ptr<Buffer> value)
                : StringScalar(std::move(value), utf8()) {}

        explicit StringScalar(std::string s) : BinaryScalar(std::move(s), utf8()) {}

        StringScalar() : StringScalar(utf8()) {}
    };

    struct TURBO_EXPORT BinaryViewScalar
            : public BaseBinaryScalar,
              private internal::array_span_fill_from_scalar_scratch_space<BinaryViewScalar> {
        using TypeClass = BinaryViewType;
        using array_span_fill_from_scalar_scratch_space =
                internal::array_span_fill_from_scalar_scratch_space<BinaryViewScalar>;

        explicit BinaryViewScalar(std::shared_ptr<DataType> type)
                : BaseBinaryScalar(std::move(type)),
                  array_span_fill_from_scalar_scratch_space(this->value) {}

        BinaryViewScalar(std::shared_ptr<Buffer> value, std::shared_ptr<DataType> type)
                : BaseBinaryScalar(std::move(value), std::move(type)),
                  array_span_fill_from_scalar_scratch_space(this->value) {}

        BinaryViewScalar(std::string s, std::shared_ptr<DataType> type)
                : BaseBinaryScalar(std::move(s), std::move(type)),
                  array_span_fill_from_scalar_scratch_space(this->value) {}

        explicit BinaryViewScalar(std::shared_ptr<Buffer> value)
                : BinaryViewScalar(std::move(value), binary_view()) {}

        explicit BinaryViewScalar(std::string s)
                : BinaryViewScalar(std::move(s), binary_view()) {}

        BinaryViewScalar() : BinaryViewScalar(binary_view()) {}

        std::string_view view() const override { return std::string_view(*this->value); }

    private:
        static void FillScratchSpace(uint8_t *scratch_space,
                                     const std::shared_ptr<Buffer> &value);

        friend ArraySpan;
        friend array_span_fill_from_scalar_scratch_space;
    };

    struct TURBO_EXPORT StringViewScalar : public BinaryViewScalar {
        using BinaryViewScalar::BinaryViewScalar;
        using TypeClass = StringViewType;

        explicit StringViewScalar(std::shared_ptr<Buffer> value)
                : StringViewScalar(std::move(value), utf8_view()) {}

        explicit StringViewScalar(std::string s)
                : BinaryViewScalar(std::move(s), utf8_view()) {}

        StringViewScalar() : StringViewScalar(utf8_view()) {}
    };

    struct TURBO_EXPORT LargeBinaryScalar
            : public BaseBinaryScalar,
              private internal::array_span_fill_from_scalar_scratch_space<LargeBinaryScalar> {
        using TypeClass = LargeBinaryType;
        using array_span_fill_from_scalar_scratch_space =
                internal::array_span_fill_from_scalar_scratch_space<LargeBinaryScalar>;

        explicit LargeBinaryScalar(std::shared_ptr<DataType> type)
                : BaseBinaryScalar(std::move(type)),
                  array_span_fill_from_scalar_scratch_space(this->value) {}

        LargeBinaryScalar(std::shared_ptr<Buffer> value, std::shared_ptr<DataType> type)
                : BaseBinaryScalar(std::move(value), std::move(type)),
                  array_span_fill_from_scalar_scratch_space(this->value) {}

        LargeBinaryScalar(std::string s, std::shared_ptr<DataType> type)
                : BaseBinaryScalar(std::move(s), std::move(type)),
                  array_span_fill_from_scalar_scratch_space(this->value) {}

        explicit LargeBinaryScalar(std::shared_ptr<Buffer> value)
                : LargeBinaryScalar(std::move(value), large_binary()) {}

        explicit LargeBinaryScalar(std::string s)
                : LargeBinaryScalar(std::move(s), large_binary()) {}

        LargeBinaryScalar() : LargeBinaryScalar(large_binary()) {}

    private:
        static void FillScratchSpace(uint8_t *scratch_space,
                                     const std::shared_ptr<Buffer> &value);

        friend ArraySpan;
        friend array_span_fill_from_scalar_scratch_space;
    };

    struct TURBO_EXPORT LargeStringScalar : public LargeBinaryScalar {
        using LargeBinaryScalar::LargeBinaryScalar;
        using TypeClass = LargeStringType;

        explicit LargeStringScalar(std::shared_ptr<Buffer> value)
                : LargeStringScalar(std::move(value), large_utf8()) {}

        explicit LargeStringScalar(std::string s)
                : LargeBinaryScalar(std::move(s), large_utf8()) {}

        LargeStringScalar() : LargeStringScalar(large_utf8()) {}
    };

    struct TURBO_EXPORT FixedSizeBinaryScalar : public BinaryScalar {
        using TypeClass = FixedSizeBinaryType;

        FixedSizeBinaryScalar(std::shared_ptr<Buffer> value, std::shared_ptr<DataType> type,
                              bool is_valid = true);

        explicit FixedSizeBinaryScalar(const std::shared_ptr<Buffer> &value,
                                       bool is_valid = true);

        explicit FixedSizeBinaryScalar(std::string s, bool is_valid = true);
    };

    template<typename T>
    struct TemporalScalar : internal::PrimitiveScalar<T> {
        using internal::PrimitiveScalar<T>::PrimitiveScalar;
        using value_type = typename internal::PrimitiveScalar<T>::value_type;

        TemporalScalar(value_type value, std::shared_ptr<DataType> type)
                : internal::PrimitiveScalar<T>(std::move(value), type) {}
    };

    template<typename T>
    struct DateScalar : public TemporalScalar<T> {
        using TemporalScalar<T>::TemporalScalar;
        using value_type = typename TemporalScalar<T>::value_type;

        explicit DateScalar(value_type value)
                : TemporalScalar<T>(std::move(value), TypeTraits<T>::type_singleton()) {}

        DateScalar() : TemporalScalar<T>(TypeTraits<T>::type_singleton()) {}
    };

    struct TURBO_EXPORT Date32Scalar : public DateScalar<Date32Type> {
        using DateScalar<Date32Type>::DateScalar;
    };

    struct TURBO_EXPORT Date64Scalar : public DateScalar<Date64Type> {
        using DateScalar<Date64Type>::DateScalar;
    };

    template<typename T>
    struct TURBO_EXPORT TimeScalar : public TemporalScalar<T> {
        using TemporalScalar<T>::TemporalScalar;

        TimeScalar(typename TemporalScalar<T>::value_type value, TimeUnit::type unit)
                : TimeScalar(std::move(value), std::make_shared<T>(unit)) {}
    };

    struct TURBO_EXPORT Time32Scalar : public TimeScalar<Time32Type> {
        using TimeScalar<Time32Type>::TimeScalar;
    };

    struct TURBO_EXPORT Time64Scalar : public TimeScalar<Time64Type> {
        using TimeScalar<Time64Type>::TimeScalar;
    };

    struct TURBO_EXPORT TimestampScalar : public TemporalScalar<TimestampType> {
        using TemporalScalar<TimestampType>::TemporalScalar;

        TimestampScalar(typename TemporalScalar<TimestampType>::value_type value,
                        TimeUnit::type unit, std::string tz = "")
                : TimestampScalar(std::move(value), timestamp(unit, std::move(tz))) {}

        static turbo::Result<TimestampScalar> FromISO8601(std::string_view iso8601,
                                                          TimeUnit::type unit);
    };

    template<typename T>
    struct IntervalScalar : public TemporalScalar<T> {
        using TemporalScalar<T>::TemporalScalar;
        using value_type = typename TemporalScalar<T>::value_type;

        explicit IntervalScalar(value_type value)
                : TemporalScalar<T>(value, TypeTraits<T>::type_singleton()) {}

        IntervalScalar() : TemporalScalar<T>(TypeTraits<T>::type_singleton()) {}
    };

    struct TURBO_EXPORT MonthIntervalScalar : public IntervalScalar<MonthIntervalType> {
        using IntervalScalar<MonthIntervalType>::IntervalScalar;
    };

    struct TURBO_EXPORT DayTimeIntervalScalar : public IntervalScalar<DayTimeIntervalType> {
        using IntervalScalar<DayTimeIntervalType>::IntervalScalar;
    };

    struct TURBO_EXPORT MonthDayNanoIntervalScalar
            : public IntervalScalar<MonthDayNanoIntervalType> {
        using IntervalScalar<MonthDayNanoIntervalType>::IntervalScalar;
    };

    struct TURBO_EXPORT DurationScalar : public TemporalScalar<DurationType> {
        using TemporalScalar<DurationType>::TemporalScalar;

        DurationScalar(typename TemporalScalar<DurationType>::value_type value,
                       TimeUnit::type unit)
                : DurationScalar(std::move(value), duration(unit)) {}

        // Convenience constructors for a DurationScalar from std::chrono::nanoseconds
        template<template<typename, typename> class StdDuration, typename Rep>
        explicit DurationScalar(StdDuration<Rep, std::nano> d)
                : DurationScalar{DurationScalar(d.count(), duration(TimeUnit::NANO))} {}

        // Convenience constructors for a DurationScalar from std::chrono::microseconds
        template<template<typename, typename> class StdDuration, typename Rep>
        explicit DurationScalar(StdDuration<Rep, std::micro> d)
                : DurationScalar{DurationScalar(d.count(), duration(TimeUnit::MICRO))} {}

        // Convenience constructors for a DurationScalar from std::chrono::milliseconds
        template<template<typename, typename> class StdDuration, typename Rep>
        explicit DurationScalar(StdDuration<Rep, std::milli> d)
                : DurationScalar{DurationScalar(d.count(), duration(TimeUnit::MILLI))} {}

        // Convenience constructors for a DurationScalar from std::chrono::seconds
        // or from units which are whole numbers of seconds
        template<template<typename, typename> class StdDuration, typename Rep, intmax_t Num>
        explicit DurationScalar(StdDuration<Rep, std::ratio<Num, 1>> d)
                : DurationScalar{DurationScalar(d.count() * Num, duration(TimeUnit::SECOND))} {}
    };

    template<typename TYPE_CLASS, typename VALUE_TYPE>
    struct TURBO_EXPORT DecimalScalar : public internal::PrimitiveScalarBase {
        using internal::PrimitiveScalarBase::PrimitiveScalarBase;
        using TypeClass = TYPE_CLASS;
        using value_type = VALUE_TYPE;

        DecimalScalar(value_type value, std::shared_ptr<DataType> type)
                : internal::PrimitiveScalarBase(std::move(type), true), value(value) {}

        const void *data() const override {
            return reinterpret_cast<const void *>(value.native_endian_bytes());
        }

        std::string_view view() const override {
            return std::string_view(reinterpret_cast<const char *>(value.native_endian_bytes()),
                                    value_type::kByteWidth);
        }

        value_type value;
    };

    struct TURBO_EXPORT Decimal128Scalar : public DecimalScalar<Decimal128Type, Decimal128> {
        using DecimalScalar::DecimalScalar;
    };

    struct TURBO_EXPORT Decimal256Scalar : public DecimalScalar<Decimal256Type, Decimal256> {
        using DecimalScalar::DecimalScalar;
    };

    struct TURBO_EXPORT BaseListScalar : public Scalar {
        using value_type = std::shared_ptr<Array>;

        BaseListScalar(std::shared_ptr<Array> value, std::shared_ptr<DataType> type,
                       bool is_valid = true);

        // The value is not supposed to be modified after construction, because subclasses have
        // a scratch space whose content need to be kept consistent with the value. It is also
        // the user of this class's responsibility to ensure that the array is not modified
        // accidentally.
        const std::shared_ptr<Array> value;
    };

    struct TURBO_EXPORT ListScalar
            : public BaseListScalar,
              private internal::array_span_fill_from_scalar_scratch_space<ListScalar> {
        using TypeClass = ListType;
        using array_span_fill_from_scalar_scratch_space =
                internal::array_span_fill_from_scalar_scratch_space<ListScalar>;

        ListScalar(std::shared_ptr<Array> value, std::shared_ptr<DataType> type,
                   bool is_valid = true)
                : BaseListScalar(std::move(value), std::move(type), is_valid),
                  array_span_fill_from_scalar_scratch_space(this->value) {}

        explicit ListScalar(std::shared_ptr<Array> value, bool is_valid = true);

    private:
        static void FillScratchSpace(uint8_t *scratch_space,
                                     const std::shared_ptr<Array> &value);

        friend ArraySpan;
        friend array_span_fill_from_scalar_scratch_space;
    };

    struct TURBO_EXPORT LargeListScalar
            : public BaseListScalar,
              private internal::array_span_fill_from_scalar_scratch_space<LargeListScalar> {
        using TypeClass = LargeListType;
        using array_span_fill_from_scalar_scratch_space =
                internal::array_span_fill_from_scalar_scratch_space<LargeListScalar>;

        LargeListScalar(std::shared_ptr<Array> value, std::shared_ptr<DataType> type,
                        bool is_valid = true)
                : BaseListScalar(std::move(value), std::move(type), is_valid),
                  array_span_fill_from_scalar_scratch_space(this->value) {}

        explicit LargeListScalar(std::shared_ptr<Array> value, bool is_valid = true);

    private:
        static void FillScratchSpace(uint8_t *scratch_space,
                                     const std::shared_ptr<Array> &value);

        friend ArraySpan;
        friend array_span_fill_from_scalar_scratch_space;
    };

    struct TURBO_EXPORT ListViewScalar
            : public BaseListScalar,
              private internal::array_span_fill_from_scalar_scratch_space<ListViewScalar> {
        using TypeClass = ListViewType;
        using array_span_fill_from_scalar_scratch_space =
                internal::array_span_fill_from_scalar_scratch_space<ListViewScalar>;

        ListViewScalar(std::shared_ptr<Array> value, std::shared_ptr<DataType> type,
                       bool is_valid = true)
                : BaseListScalar(std::move(value), std::move(type), is_valid),
                  array_span_fill_from_scalar_scratch_space(this->value) {}

        explicit ListViewScalar(std::shared_ptr<Array> value, bool is_valid = true);

    private:
        static void FillScratchSpace(uint8_t *scratch_space,
                                     const std::shared_ptr<Array> &value);

        friend ArraySpan;
        friend array_span_fill_from_scalar_scratch_space;
    };

    struct TURBO_EXPORT LargeListViewScalar
            : public BaseListScalar,
              private internal::array_span_fill_from_scalar_scratch_space<LargeListViewScalar> {
        using TypeClass = LargeListViewType;
        using array_span_fill_from_scalar_scratch_space =
                internal::array_span_fill_from_scalar_scratch_space<LargeListViewScalar>;

        LargeListViewScalar(std::shared_ptr<Array> value, std::shared_ptr<DataType> type,
                            bool is_valid = true)
                : BaseListScalar(std::move(value), std::move(type), is_valid),
                  array_span_fill_from_scalar_scratch_space(this->value) {}

        explicit LargeListViewScalar(std::shared_ptr<Array> value, bool is_valid = true);

    private:
        static void FillScratchSpace(uint8_t *scratch_space,
                                     const std::shared_ptr<Array> &value);

        friend ArraySpan;
        friend array_span_fill_from_scalar_scratch_space;
    };

    struct TURBO_EXPORT MapScalar
            : public BaseListScalar,
              private internal::array_span_fill_from_scalar_scratch_space<MapScalar> {
        using TypeClass = MapType;
        using array_span_fill_from_scalar_scratch_space =
                internal::array_span_fill_from_scalar_scratch_space<MapScalar>;

        MapScalar(std::shared_ptr<Array> value, std::shared_ptr<DataType> type,
                  bool is_valid = true)
                : BaseListScalar(std::move(value), std::move(type), is_valid),
                  array_span_fill_from_scalar_scratch_space(this->value) {}

        explicit MapScalar(std::shared_ptr<Array> value, bool is_valid = true);

    private:
        static void FillScratchSpace(uint8_t *scratch_space,
                                     const std::shared_ptr<Array> &value);

        friend ArraySpan;
        friend array_span_fill_from_scalar_scratch_space;
    };

    struct TURBO_EXPORT FixedSizeListScalar : public BaseListScalar {
        using TypeClass = FixedSizeListType;

        FixedSizeListScalar(std::shared_ptr<Array> value, std::shared_ptr<DataType> type,
                            bool is_valid = true);

        explicit FixedSizeListScalar(std::shared_ptr<Array> value, bool is_valid = true);
    };

    struct TURBO_EXPORT StructScalar : public Scalar {
        using TypeClass = StructType;
        using value_type = std::vector<std::shared_ptr<Scalar>>;

        ScalarVector value;

        turbo::Result<std::shared_ptr<Scalar>> field(FieldRef ref) const;

        StructScalar(value_type value, std::shared_ptr<DataType> type, bool is_valid = true)
                : Scalar(std::move(type), is_valid), value(std::move(value)) {}

        static turbo::Result<std::shared_ptr<StructScalar>> create(value_type value,
                                                                 std::vector<std::string> field_names);
    };

    struct TURBO_EXPORT UnionScalar : public Scalar {
        // The type code is not supposed to be modified after construction, because the scratch
        // space's content need to be kept consistent with it.
        const int8_t type_code;

        virtual const std::shared_ptr<Scalar> &child_value() const = 0;

    protected:
        UnionScalar(std::shared_ptr<DataType> type, int8_t type_code, bool is_valid)
                : Scalar(std::move(type), is_valid), type_code(type_code) {}

        struct UnionScratchSpace {
            alignas(int64_t) int8_t type_code;
            alignas(int64_t) uint8_t offsets[sizeof(int32_t) * 2];
        };
        static_assert(sizeof(UnionScratchSpace) <= internal::kScalarScratchSpaceSize);

        friend ArraySpan;
    };

    struct TURBO_EXPORT SparseUnionScalar
            : public UnionScalar,
              private internal::array_span_fill_from_scalar_scratch_space<SparseUnionScalar> {
        using TypeClass = SparseUnionType;
        using array_span_fill_from_scalar_scratch_space =
                internal::array_span_fill_from_scalar_scratch_space<SparseUnionScalar>;

        // Even though only one of the union values is relevant for this scalar, we
        // nonetheless construct a vector of scalars, one per union value, to have
        // enough data to reconstruct a valid ArraySpan of length 1 from this scalar
        using value_type = std::vector<std::shared_ptr<Scalar>>;
        // The value is not supposed to be modified after construction, because the scratch
        // space's content need to be kept consistent with the value. It is also the user of
        // this class's responsibility to ensure that the scalars of the vector is not modified
        // to accidentally.
        const value_type value;

        // The value index corresponding to the active type code
        int child_id;

        SparseUnionScalar(value_type value, int8_t type_code, std::shared_ptr<DataType> type);

        const std::shared_ptr<Scalar> &child_value() const override {
            return this->value[this->child_id];
        }

        /// \brief Construct a SparseUnionScalar from a single value, versus having
        /// to construct a vector of scalars
        static std::shared_ptr<Scalar> FromValue(std::shared_ptr<Scalar> value, int field_index,
                                                 std::shared_ptr<DataType> type);

    private:
        static void FillScratchSpace(uint8_t *scratch_space, int8_t type_code);

        friend ArraySpan;
        friend array_span_fill_from_scalar_scratch_space;
    };

    struct TURBO_EXPORT DenseUnionScalar
            : public UnionScalar,
              private internal::array_span_fill_from_scalar_scratch_space<DenseUnionScalar> {
        using TypeClass = DenseUnionType;
        using array_span_fill_from_scalar_scratch_space =
                internal::array_span_fill_from_scalar_scratch_space<DenseUnionScalar>;

        // For DenseUnionScalar, we can make a valid ArraySpan of length 1 from this
        // scalar
        using value_type = std::shared_ptr<Scalar>;
        // The value is not supposed to be modified after construction, because the scratch
        // space's content need to be kept consistent with the value. It is also the user of
        // this class's responsibility to ensure that the elements of the vector is not modified
        // accidentally.
        const value_type value;

        const std::shared_ptr<Scalar> &child_value() const override { return this->value; }

        DenseUnionScalar(value_type value, int8_t type_code, std::shared_ptr<DataType> type)
                : UnionScalar(std::move(type), type_code, value->is_valid),
                  array_span_fill_from_scalar_scratch_space(type_code),
                  value(std::move(value)) {}

    private:
        static void FillScratchSpace(uint8_t *scratch_space, int8_t type_code);

        friend ArraySpan;
        friend array_span_fill_from_scalar_scratch_space;
    };

    struct TURBO_EXPORT RunEndEncodedScalar
            : public Scalar,
              private internal::array_span_fill_from_scalar_scratch_space<RunEndEncodedScalar> {
        using TypeClass = RunEndEncodedType;
        using value_type = std::shared_ptr<Scalar>;
        using array_span_fill_from_scalar_scratch_space =
                internal::array_span_fill_from_scalar_scratch_space<RunEndEncodedScalar>;

        // The value is not supposed to be modified after construction, because the scratch
        // space's content need to be kept consistent with the value. It is also the user of
        // this class's responsibility to ensure that the wrapped scalar is not modified
        // accidentally.
        const value_type value;

        RunEndEncodedScalar(std::shared_ptr<Scalar> value, std::shared_ptr<DataType> type);

        /// \brief Constructs a NULL RunEndEncodedScalar
        explicit RunEndEncodedScalar(const std::shared_ptr<DataType> &type);

        ~RunEndEncodedScalar() override;

        const std::shared_ptr<DataType> &run_end_type() const {
            return ree_type().run_end_type();
        }

        const std::shared_ptr<DataType> &get_value_type() const { return ree_type().get_value_type(); }

    private:
        const TypeClass &ree_type() const { return turbo::checked_cast<TypeClass &>(*type); }

        static void FillScratchSpace(uint8_t *scratch_space, const DataType &type);

        friend ArraySpan;
        friend array_span_fill_from_scalar_scratch_space;
    };

    /// \brief A Scalar value for DictionaryType
    ///
    /// `is_valid` denotes the validity of the `index`, regardless of
    /// the corresponding value in the `dictionary`.
    struct TURBO_EXPORT DictionaryScalar : public internal::PrimitiveScalarBase {
        using TypeClass = DictionaryType;
        struct value_type {
            std::shared_ptr<Scalar> index;
            std::shared_ptr<Array> dictionary;
        } value;

        explicit DictionaryScalar(std::shared_ptr<DataType> type);

        DictionaryScalar(value_type value, std::shared_ptr<DataType> type, bool is_valid = true)
                : internal::PrimitiveScalarBase(std::move(type), is_valid),
                  value(std::move(value)) {}

        static std::shared_ptr<DictionaryScalar> create(std::shared_ptr<Scalar> index,
                                                      std::shared_ptr<Array> dict);

        turbo::Result<std::shared_ptr<Scalar>> GetEncodedValue() const;

        const void *data() const override {
            return turbo::checked_cast<internal::PrimitiveScalarBase &>(*value.index).data();
        }

        std::string_view view() const override {
            return turbo::checked_cast<const internal::PrimitiveScalarBase &>(*value.index)
                    .view();
        }
    };

    /// \brief A Scalar value for ExtensionType
    ///
    /// The value is the underlying storage scalar.
    /// `is_valid` must only be true if `value` is non-null and `value->is_valid` is true
    struct TURBO_EXPORT ExtensionScalar : public Scalar {
        using TypeClass = ExtensionType;
        using value_type = std::shared_ptr<Scalar>;

        ExtensionScalar(std::shared_ptr<Scalar> storage, std::shared_ptr<DataType> type,
                        bool is_valid = true)
                : Scalar(std::move(type), is_valid), value(std::move(storage)) {}

        template<typename Storage,
                typename = enable_if_t<std::is_base_of<Scalar, Storage>::value>>
        ExtensionScalar(Storage &&storage, std::shared_ptr<DataType> type, bool is_valid = true)
                : ExtensionScalar(std::make_shared<Storage>(std::move(storage)), std::move(type),
                                  is_valid) {}

        std::shared_ptr<Scalar> value;
    };

    /// @}

    namespace internal {

        inline turbo::Status CheckBufferLength(...) { return turbo::OkStatus(); }

        TURBO_EXPORT turbo::Status CheckBufferLength(const FixedSizeBinaryType *t,
                                                     const std::shared_ptr<Buffer> *b);

    }  // namespace internal

    template<typename ValueRef>
    struct MakeScalarImpl;

    /// \defgroup scalar-factories Scalar factory functions
    ///
    /// @{

    /// \brief Scalar factory for null scalars
    TURBO_EXPORT
    std::shared_ptr<Scalar> MakeNullScalar(std::shared_ptr<DataType> type);

    /// \brief Scalar factory for non-null scalars
    template<typename Value>
    turbo::Result<std::shared_ptr<Scalar>> MakeScalar(std::shared_ptr<DataType> type,
                                                      Value &&value) {
        return MakeScalarImpl<Value &&>{type, std::forward<Value>(value), nullptr}.finish();
    }

    /// \brief Type-inferring scalar factory for non-null scalars
    ///
    /// Construct a Scalar instance with a DataType determined by the input C++ type.
    /// (for example Int8Scalar for a int8_t input).
    /// Only non-parametric primitive types and String are supported.
    template<typename Value, typename Traits = CTypeTraits<typename std::decay<Value>::type>,
            typename ScalarType = typename Traits::ScalarType,
            typename Enable = decltype(ScalarType(std::declval<Value>(),
                                                  Traits::type_singleton()))>
    std::shared_ptr<Scalar> MakeScalar(Value value) {
        return std::make_shared<ScalarType>(std::move(value), Traits::type_singleton());
    }

    inline std::shared_ptr<Scalar> MakeScalar(std::string value) {
        return std::make_shared<StringScalar>(std::move(value));
    }

    inline std::shared_ptr<Scalar> MakeScalar(const std::shared_ptr<Scalar> &scalar) {
        return scalar;
    }
    /// @}

    template<typename ValueRef>
    struct MakeScalarImpl {
        template<typename T, typename ScalarType = typename TypeTraits<T>::ScalarType,
                typename VT = typename ScalarType::value_type,
                typename Enable = typename std::enable_if<
                        std::is_constructible<ScalarType, VT,
                                std::shared_ptr<DataType>>::value &&
                        std::is_convertible<ValueRef, VT>::value>::type>
        turbo::Status Visit(const T &t) {
            TURBO_RETURN_NOT_OK(internal::CheckBufferLength(&t, &value_));
            // `static_cast<ValueRef>` makes a rvalue if ValueRef is `VT&&`
            out_ = std::make_shared<ScalarType>(
                    static_cast<VT>(static_cast<ValueRef>(value_)), std::move(type_));
            return turbo::OkStatus();
        }

        turbo::Status Visit(const ExtensionType &t) {
            TURBO_MOVE_OR_RAISE(auto storage,
                                MakeScalar(t.storage_type(), static_cast<ValueRef>(value_)));
            out_ = std::make_shared<ExtensionScalar>(std::move(storage), type_);
            return turbo::OkStatus();
        }

        // Enable constructing string/binary scalars (but not decimal, etc) from std::string
        template<typename T>
        enable_if_t<
                std::is_same<typename std::remove_reference<ValueRef>::type, std::string>::value &&
                (is_base_binary_type<T>::value || std::is_same<T, FixedSizeBinaryType>::value),
                turbo::Status>
        Visit(const T &t) {
            using ScalarType = typename TypeTraits<T>::ScalarType;
            out_ = std::make_shared<ScalarType>(Buffer::from_string(std::move(value_)),
                                                std::move(type_));
            return turbo::OkStatus();
        }

        turbo::Status Visit(const DataType &t) {
            return turbo::unimplemented_error("constructing scalars of type ", t,
                                                 " from unboxed values");
        }

        turbo::Result<std::shared_ptr<Scalar>> finish() &&{
            TURBO_RETURN_NOT_OK(visit_type_inline(*type_, this));
            return std::move(out_);
        }

        std::shared_ptr<DataType> type_;
        ValueRef value_;
        std::shared_ptr<Scalar> out_;
    };

}  // namespace nebula
